
#include "Debug.h"
#include "antx.h"
#include "clientAcl.h"
#include "Helper.h"

#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <signal.h>
#include <sys/resource.h>
#include <sys/uio.h>
#include <pthread.h>
/* some POSIX systems need the following definition
 * to get mlockall flags out of sys/mman.h.  */
#ifndef _P1003_1B_VISIBLE
#define _P1003_1B_VISIBLE
#endif
/* need this to get IOV_MAX on some platforms. */
#ifndef __need_IOV_MAX
#define __need_IOV_MAX
#endif
#include <pwd.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <assert.h>
#include <limits.h>

/* OpenBSD has a malloc.h, but warns to use stdlib.h instead */
#ifndef __OpenBSD__
#include <malloc.h>
#endif

/* FreeBSD 4.x doesn't have IOV_MAX exposed. */
#ifndef IOV_MAX
#if defined(__FreeBSD__) || defined(__APPLE__)
# define IOV_MAX 1024
#endif
#endif

#include "exmonwaiter.h"

int zexmon_port = 0;
int report_port = 0;
int sock_fd = 0;
static int uptime  = 0;

/*
 * forward declarations
 */
static void drive_machine(conn *c);
static int new_socket(struct addrinfo *ai);
static int server_socket(const int port, const bool is_udp);
static int try_read_command(conn *c);
static int try_read_network(conn *c);
static int try_read_udp(conn *c);

/* defaults */
static void settings_init(void);

/* event handling, network IO */
static void event_handler(const int fd, const short which, void *arg);
static void conn_close(conn *c);
static void conn_init(void);
static bool update_event(conn *c, const int new_flags);
static void complete_nread(conn *c);
static void process_command(conn *c, char *command);
static int transmit(conn *c);
static int ensure_iov_space(conn *c);
static int add_iov(conn *c, const void *buf, int len);
static int add_msghdr(conn *c);

static void conn_wbuf_shrink(conn *c) ;
 
/* time handling */
static void conn_free(conn *c);

/** exported globals **/
struct settings settings;
static conn *listen_conn = NULL;
static struct event_base *main_base;
static const string HELPERCONF = "helper.conf";

#define TRANSMIT_COMPLETE   0
#define TRANSMIT_INCOMPLETE 1
#define TRANSMIT_SOFT_ERROR 2
#define TRANSMIT_HARD_ERROR 3

typedef struct token_s {
    char *value;
    size_t length;
} token_t;

#define COMMAND_TOKEN      0
#define SUBCOMMAND_TOKEN   1
#define KEY_TOKEN          1
#define KEY_MAX_LENGTH     250
#define MAX_TOKENS         100 
//#define MAX_TOKENS         8

/*
 * Tokenize the command string by replacing whitespace with '\0' and update
 * the token array tokens with pointer to start of each token and length.
 * Returns total number of tokens.  The last valid token is the terminal
 * token (value points to the first unprocessed character of the string and
 * length zero).
 *
 * Usage example:
 *
 *  while(tokenize_command(command, ncommand, tokens, max_tokens) > 0) {
 *      for(int ix = 0; tokens[ix].length != 0; ix++) {
 *          ...
 *      }
 *      ncommand = tokens[ix].value - command;
 *      command  = tokens[ix].value;
 *   }
 */
static size_t tokenize_command(char *command, token_t *tokens, const size_t max_tokens) {
    char *s, *e;
    size_t ntokens = 0;

    assert(command != NULL && tokens != NULL && max_tokens > 1);

    for (s = e = command; ntokens < max_tokens - 1; ++e) {
        if (*e == ' ') {
            if (s != e) {
                tokens[ntokens].value = s;
                tokens[ntokens].length = e - s;
                ntokens++;
                *e = '\0';
            }
            s = e + 1;
        }
        else if (*e == '\0') {
            if (s != e) {
                tokens[ntokens].value = s;
                tokens[ntokens].length = e - s;
                ntokens++;
            }

            break; /* string end */
        }
    }

    /*
     * If we scanned the whole string, the terminal value pointer is null,
     * otherwise it is the first unprocessed character.
     */
    tokens[ntokens].value =  *e == '\0' ? NULL : e;
    tokens[ntokens].length = 0;
    ntokens++;

    return ntokens;
}


static void settings_init(void) {
    settings.access=0700;
    settings.port = 11211;
    settings.udpport = 0;  //11211;
    /* By default this string should be NULL for getaddrinfo() */
    settings.inter = NULL;
    settings.maxconns = 1024;         /* to limit connections-related memory to about 5MB */
    settings.socketpath = NULL;       /* by default, not using a unix socket */
    settings.num_threads = 4;
    settings.num_threads++;  /* N workers + 1 dispatcher */
    settings.detail_enabled = 0;
    settings.reqs_per_event = 20;
    settings.backlog = 1024;
    settings.authfile = NULL;
    settings.helper = (char *)HELPERCONF.c_str();
    settings.timer = 0;
}

/*
 * Adds a message header to a connection.
 * Returns 0 on success, -1 on out-of-memory.
 */
static int add_msghdr(conn *c)
{
    struct msghdr *msg;
    assert(c != NULL);
    if (c->msgsize == c->msgused) {
        msg = (struct msghdr * )realloc(c->msglist, c->msgsize * 2 * sizeof(struct msghdr));
        if (! msg)
            return -1;
        c->msglist = msg;
        c->msgsize *= 2;
    }

    msg = c->msglist + c->msgused;
    /* this wipes msg_iovlen, msg_control, msg_controllen, and
       msg_flags, the last 3 of which aren't defined on solaris: */
    memset(msg, 0, sizeof(struct msghdr));
    msg->msg_iov = &c->iov[c->iovused];
    if (c->request_addr_size > 0) {
        msg->msg_name = &c->request_addr;
        msg->msg_namelen = c->request_addr_size;
    }
    c->msgbytes = 0;
    c->msgused++;
    if (c->udp) {
        /* Leave room for the UDP header, which we'll fill in later. */
        return add_iov(c, NULL, UDP_HEADER_SIZE);
    }

    return 0;
}


/*
 * Free list management for connections.
 */

static conn **freeconns;
static int freetotal;
static int freecurr;

static void conn_init(void) {
    freetotal = 200;
    freecurr = 0;
    if ((freeconns = (conn **)malloc(sizeof(conn *) * freetotal)) == NULL) {
        fprintf(stderr, "malloc()\n");
    }
    return;
}

/*
 * Returns a connection from the freelist, if any. Should call this using
 * conn_from_freelist() for thread safety.
 */
conn *do_conn_from_freelist() {
    conn *c;
    if (freecurr > 0) {
        c = freeconns[--freecurr];
    } else {
        c = NULL;
    }
DEBUG2("freecurr = ", freecurr);
    return c;
}

/*
 * Adds a connection to the freelist. 0 = success. Should call this using
 * conn_add_to_freelist() for thread safety.
 */
bool do_conn_add_to_freelist(conn *c) {
    if (freecurr < freetotal) {
        freeconns[freecurr++] = c;
DEBUG2("current freecurr = ", freecurr);
        return false;
    } else {
        /* try to enlarge free connections array */
DEBUG1("freetotal is too small, realloc *2" );
        conn **new_freeconns = (conn **)realloc(freeconns, sizeof(conn *) * freetotal * 2);
        if (new_freeconns) {
            freetotal *= 2;
            freeconns = new_freeconns;
            freeconns[freecurr++] = c;
            return false;
        }
    }
    return true;
}

conn *conn_new(const int sfd, const int init_state, const int event_flags,
                const int read_buffer_size, const bool is_udp, struct event_base *base) {
    conn *c = conn_from_freelist();

    if (NULL == c) {
        if (!(c = (conn *)calloc(1, sizeof(conn)))) {
            fprintf(stderr, "calloc()\n");
            return NULL;
        }

        c->rbuf = c->wbuf = 0;
        c->iov = 0;
        c->msglist = 0;
        c->hdrbuf = 0;

        c->rsize = read_buffer_size;
        c->wsize = DATA_BUFFER_SIZE;
        c->iovsize = IOV_LIST_INITIAL;
        c->msgsize = MSG_LIST_INITIAL;
        c->hdrsize = 0;

        c->rbuf = (char *)malloc((size_t)c->rsize);
        c->wbuf = (char *)malloc((size_t)c->wsize);
        c->iov = (struct iovec *)malloc(sizeof(struct iovec) * c->iovsize);
        c->msglist = (struct msghdr *)malloc(sizeof(struct msghdr) * c->msgsize);

        if (c->rbuf == 0 || c->wbuf == 0 || c->iov == 0 || c->msglist == 0 ) {
            conn_free(c);
            fprintf(stderr, "malloc()\n");
            return NULL;
        }

    }

    if (init_state == conn_listening)
	    DEBUG4(" server listening port =  ", settings.port, "; fd = ", sfd );
    else if (is_udp)
	    DEBUG4(" server listening (udp) port =  ", settings.udpport, "; fd = ", sfd );
    else
	    DEBUG2("new client connection fd = ", sfd);

    c->sfd = sfd;
    c->udp = is_udp;
    c->state = init_state;
    c->rlbytes = 0;
    c->rbytes = c->wbytes = 0;
    c->wcurr = c->wbuf;
    c->rcurr = c->rbuf;
    c->ritem = 0;
    c->offset = 0; 
    c->iovused = 0;
    c->msgcurr = 0;
    c->msgused = 0;
    c->key = NULL;
    c->nkey = 0;
    c->flags = 0;
    c->expire = 0;

    c->write_and_go = conn_read;
    c->write_and_free = 0;

    c->noreply = false;

    event_set(&c->event, sfd, event_flags, event_handler, (void *)c);
    event_base_set(base, &c->event);
    c->ev_flags = event_flags;

    if (event_add(&c->event, 0) == -1) {
        if (conn_add_to_freelist(c)) {
            conn_free(c);
        }
        perror("event_add");
        return NULL;
    }
    return c;
}

static void conn_cleanup(conn *c) {
    assert(c != NULL);
    if (c->write_and_free) {
        free(c->write_and_free);
        c->write_and_free = 0;
    }
}

/*
 * Frees a connection.
 */
void conn_free(conn *c) {
    if (c) {
        if (c->hdrbuf)
            free(c->hdrbuf);
        if (c->msglist)
            free(c->msglist);
        if (c->rbuf)
            free(c->rbuf);
        if (c->wbuf)
            free(c->wbuf);
        if (c->iov)
            free(c->iov);
        free(c);
    }
}

static void conn_close(conn *c) {
    assert(c != NULL);
    /* delete the event, the socket and the conn */
    event_del(&c->event);
    DEBUG2("connection closed fd = ", c->sfd);
    close(c->sfd);
    accept_new_conns(true);
    conn_cleanup(c);
    /* if the connection has big buffers, just free it */
    if (c->rsize > READ_BUFFER_HIGHWAT || conn_add_to_freelist(c)) {
        conn_free(c);
    }

    return;
}


/*
 * Shrinks a connection's buffers if they're too big.  This prevents
 * periodic large "get" requests from permanently chewing lots of server
 * memory.
 *
 * This should only be called in between requests since it can wipe output
 * buffers!
 */
static void conn_shrink(conn *c) {
	assert(c != NULL);
	if (c->udp)
		return;
	if (c->rsize > READ_BUFFER_HIGHWAT && c->rbytes < DATA_BUFFER_SIZE) {
		char *newbuf;
		if (c->rcurr != c->rbuf && c->rbytes > 0)
			memmove(c->rbuf, c->rcurr, (size_t)c->rbytes);
		newbuf = (char *)realloc((void *)c->rbuf, DATA_BUFFER_SIZE);
		if (newbuf) {
			c->rbuf = newbuf;
			c->rsize = DATA_BUFFER_SIZE;
		}
		/* TODO check other branch... */
		c->rcurr = c->rbuf;
	}

    if (c->msgsize > MSG_LIST_HIGHWAT) {
        struct msghdr *newbuf = (struct msghdr *) realloc((void *)c->msglist, MSG_LIST_INITIAL * sizeof(c->msglist[0]));
        if (newbuf) {
            c->msglist = newbuf;
            c->msgsize = MSG_LIST_INITIAL;
        }
    /* TODO check error condition? */
    }

    if (c->iovsize > IOV_LIST_HIGHWAT) {
        struct iovec *newbuf = (struct iovec *) realloc((void *)c->iov, IOV_LIST_INITIAL * sizeof(c->iov[0]));
        if (newbuf) {
            c->iov = newbuf;
            c->iovsize = IOV_LIST_INITIAL;
        }
    /* TODO check return value */
    }
}

static void conn_wbuf_shrink(conn *c) {
	assert(c != NULL);
	if (c->wsize >  READ_BUFFER_HIGHWAT ) {
		char *newbuf;
               
		int curr = c->wcurr - c->wbuf ;
		if (c->wcurr != c->wbuf && c->wbytes > curr )
			memmove(c->wbuf, c->wcurr, (size_t)(c->wbytes - (c->wcurr - c->wbuf)));
                c->wbytes -= curr; 
		if(c->wbytes < DATA_BUFFER_SIZE)
		{
			newbuf = (char *)realloc((void *)c->wbuf, DATA_BUFFER_SIZE);
			if (newbuf) {
				c->wbuf = newbuf;
				c->wsize = DATA_BUFFER_SIZE;
			}
		}
		/* TODO check other branch... */
		c->wcurr = c->wbuf;
	}
}

/*
 * Sets a connection's current state in the state machine. Any special
 * processing that needs to happen on certain state transitions can
 * happen here.
 */
static void conn_set_state(conn *c, int state) {
    assert(c != NULL);

    if (state != c->state) {
        if (state == conn_read) {
            conn_shrink(c);
            // assoc_move_next_bucket();
        }
        c->state = state;
        if (state == conn_write) {
          //   MEMCACHED_PROCESS_COMMAND_END(c->sfd, c->wbuf, c->wbytes);
        }
    }
}


/*
 * Ensures that there is room for another struct iovec in a connection's
 * iov list.
 *
 * Returns 0 on success, -1 on out-of-memory.
 */
static int ensure_iov_space(conn *c) {
    assert(c != NULL);

    if (c->iovused >= c->iovsize) {
        int i, iovnum;
        struct iovec *new_iov = (struct iovec *)realloc(c->iov,
                                (c->iovsize * 2) * sizeof(struct iovec));
        if (! new_iov)
            return -1;
        c->iov = new_iov;
        c->iovsize *= 2;

        /* Point all the msghdr structures at the new list. */
        for (i = 0, iovnum = 0; i < c->msgused; i++) {
            c->msglist[i].msg_iov = &c->iov[iovnum];
            iovnum += c->msglist[i].msg_iovlen;
        }
    }

    return 0;
}


/*
 * Adds data to the list of pending data that will be written out to a
 * connection.
 *
 * Returns 0 on success, -1 on out-of-memory.
 */

static int add_iov(conn *c, const void *buf, int len) {
    struct msghdr *m;
    int leftover;
    bool limit_to_mtu;

    assert(c != NULL);

    do {
        m = &c->msglist[c->msgused - 1];
        /*
         * Limit UDP packets, and the first payloads of TCP replies, to
         * UDP_MAX_PAYLOAD_SIZE bytes.
         */
        limit_to_mtu = c->udp || (1 == c->msgused);

        /* We may need to start a new msghdr if this one is full. */
        if (m->msg_iovlen == IOV_MAX ||
            (limit_to_mtu && c->msgbytes >= UDP_MAX_PAYLOAD_SIZE)) {
            add_msghdr(c);
            m = &c->msglist[c->msgused - 1];
        }

        if (ensure_iov_space(c) != 0)
            return -1;

        /* If the fragment is too big to fit in the datagram, split it up */
        if (limit_to_mtu && len + c->msgbytes > UDP_MAX_PAYLOAD_SIZE) {
            leftover = len + c->msgbytes - UDP_MAX_PAYLOAD_SIZE;
            len -= leftover;
        } else {
            leftover = 0;
        }

        m = &c->msglist[c->msgused - 1];
        m->msg_iov[m->msg_iovlen].iov_base = (void *)buf;
        m->msg_iov[m->msg_iovlen].iov_len = len;

        c->msgbytes += len;
        c->iovused++;
        m->msg_iovlen++;

        buf = ((char *)buf) + len;
        len = leftover;
    } while (leftover > 0);

    return 0;
}


/*
 * Constructs a set of UDP headers and attaches them to the outgoing messages.
 */
static int build_udp_headers(conn *c) {
    int i;
    unsigned char *hdr;

    assert(c != NULL);

    if (c->msgused > c->hdrsize) {
        void *new_hdrbuf;
        if (c->hdrbuf)
            new_hdrbuf = realloc(c->hdrbuf, c->msgused * 2 * UDP_HEADER_SIZE);
        else
            new_hdrbuf = malloc(c->msgused * 2 * UDP_HEADER_SIZE);
        if (! new_hdrbuf)
            return -1;
        c->hdrbuf = (unsigned char *)new_hdrbuf;
        c->hdrsize = c->msgused * 2;
    }

    hdr = c->hdrbuf;
    for (i = 0; i < c->msgused; i++) {
        c->msglist[i].msg_iov[0].iov_base = hdr;
        c->msglist[i].msg_iov[0].iov_len = UDP_HEADER_SIZE;
        *hdr++ = c->request_id / 256;
        *hdr++ = c->request_id % 256;
        *hdr++ = i / 256;
        *hdr++ = i % 256;
        *hdr++ = c->msgused / 256;
        *hdr++ = c->msgused % 256;
        *hdr++ = 0;
        *hdr++ = 0;
//        assert((void *) hdr == (void *)c->msglist[i].msg_iov[0].iov_base + UDP_HEADER_SIZE);
    }

    return 0;
}


static void out_string(conn *c, const char *str) {
    size_t len;

    assert(c != NULL);

    if (c->noreply) {

        c->noreply = false;
        conn_set_state(c, conn_read);
        return;
    }

    len = strlen(str);
    if ((len + 2) > (size_t) c->wsize) {
        /* ought to be always enough. just fail for simplicity */
        str = "SERVER_ERROR output line too long";
        len = strlen(str);
    }

    memcpy(c->wbuf, str, len);
    memcpy(c->wbuf + len, "\r\n", 2);
    c->wbytes = len + 2;
    c->wcurr = c->wbuf;

    conn_set_state(c, conn_write);
    c->write_and_go = conn_read;
    return;
}

/*
 * we get here after reading the value in set/add/replace commands. The command
 * has been stored in c->item_comm, and the item is ready in c->item.
 */

static void complete_nread(conn *c) {
	assert(c != NULL);
	assert( c->vlen <= c->rbytes);
	/* ritem指向数据区起点,  在process_update_command 中被设置 */
	if (strncmp( c->ritem + c->vlen - 2, "\r\n", 2) != 0) {
		out_string(c, "CLIENT_ERROR bad data chunk");
	} else {
		Helper *p = Helper::Instance();
DEBUG2("cmd = ", c->key);
		if(p != NULL &&  p->Update(c->key, c->nkey, c->flags, c->expire, (const char *) c->ritem, c->vlen - 2) == 0)
			out_string(c, "STORED");
		else
			out_string(c, "SERVER STORE ERROR");
	}
	assert(c->rbytes >= c->vlen);
	// 此时rbytes应该是从DATA区开始到结束的数据长度！这里是把多读入的数据向前移动到，丢弃已经处理的数据! 
	int leftlen =  c->rbytes - c->vlen;    
	if (leftlen > 0 ) {
		memmove(c->rbuf, c->ritem + c->vlen, leftlen);
	}
	c->rcurr = c->rbuf;
	c->rbytes = leftlen;
}

/* set up a connection to write a buffer then free it, used for stats */
static void write_and_free(conn *c, char *buf, int bytes) {
    if (buf) {
        c->write_and_free = buf;
        c->wcurr = buf;
        c->wbytes = bytes;
        conn_set_state(c, conn_write);
        c->write_and_go = conn_read;
    } else {
        out_string(c, "SERVER_ERROR out of memory writing stats");
    }
}

static inline void set_noreply_maybe(conn *c, token_t *tokens, size_t ntokens)
{
    int noreply_index = ntokens - 2;

    /*
      NOTE: this function is not the first place where we are going to
      send the reply.  We could send it instead from process_command()
      if the request line has wrong number of tokens.  However parsing
      malformed line for "noreply" option is not reliable anyway, so
      it can't be helped.
    */
    if (tokens[noreply_index].value
        && strcmp(tokens[noreply_index].value, "noreply") == 0) {
        c->noreply = true;
    }
}

inline static void process_stats_detail(conn *c, const char *command) {
    assert(c != NULL);

    if (strcmp(command, "on") == 0) {
        settings.detail_enabled = 1;
        out_string(c, "OK");
    }
    else if (strcmp(command, "off") == 0) {
        settings.detail_enabled = 0;
        out_string(c, "OK");
    }
    else {
        out_string(c, "CLIENT_ERROR usage: stats detail on|off|dump");
    }
}

static void process_stat(conn *c, token_t *tokens, const size_t ntokens) {
    char *command;
    char *subcommand;

    assert(c != NULL);

    if(ntokens < 2) {
        out_string(c, "CLIENT_ERROR bad command line");
        return;
    }

    command = tokens[COMMAND_TOKEN].value;

    if (ntokens == 2 && strcmp(command, "stats") == 0) {
        char temp[2048];
        pid_t pid = getpid();
        char *pos = temp;
	Helper *p = Helper::Instance();
        string src;
	if(p!=NULL)
            src = p->Stats(NULL);
         
#ifndef WIN32
        struct rusage usage;
        getrusage(RUSAGE_SELF, &usage);
#endif /* !WIN32 */

        pos += sprintf(pos, "STAT pid %u\r\n", pid);
        pos += sprintf(pos, "STAT version " VERSION "\r\n");
        pos += sprintf(pos, "STAT uptime %u\r\n", uptime);

        pos += sprintf(pos, "STAT pointer_size %d\r\n", 8 * sizeof(void *));
#ifndef WIN32
        pos += sprintf(pos, "STAT rusage_user %ld.%06ld\r\n", usage.ru_utime.tv_sec, usage.ru_utime.tv_usec);
        pos += sprintf(pos, "STAT rusage_system %ld.%06ld\r\n", usage.ru_stime.tv_sec, usage.ru_stime.tv_usec);
#endif /* !WIN32 */
        pos += sprintf(pos, "STAT threads %u\r\n", settings.num_threads);

        pos += sprintf(pos,"%s", src.c_str());

        pos += sprintf(pos, "END");
        out_string(c, temp);
        return;
    }
DEBUG1("ERROR ?? ");
    subcommand = tokens[SUBCOMMAND_TOKEN].value;

    if (strcmp(subcommand, "reset") == 0) {
       // stats_reset();
        out_string(c, "RESET");
        return;
    }

    out_string(c, "ERROR");
}

/* ntokens is overwritten here... shrug.. */
static inline void process_gets_command(conn *c, token_t *tokens, size_t ntokens, bool return_cas) {
	assert(c != NULL);
        c->wbytes = 0;
        c->wcurr = c->wbuf;
        map <int, string > keys;
        int result_idx = (int)ntokens - 2;
	for(int i= 1; i <= result_idx ; i++)
        {
               string sk = tokens[i].value;
               keys[i] = sk;
	       DEBUG4("tokens[", i, "] = ", tokens[i].value);
        } 
        Helper *p = Helper::Instance();        
	int iret = -1; 
	if( p != NULL )
	{
DEBUG2( pthread_self(), " begin to Process");
		iret = p->Process(keys, c);
DEBUG2( pthread_self(), " end Process");
	}
	if( c->wbytes >(c->wcurr - c->wbuf))
	{
		char rbuf[256]; /*
		for(idx  =1; idx <(int) ntokens - 2; idx ++)
		{ 
			int hlen = sprintf(rbuf, " 0 %d\r\n",1);
			respH[nt].header.append("VALUE ");
			respH[nt].header.append( keys[ idx ].c_str(), keys[ idx ].c_str());
			respH[nt].header.append(rbuf, hlen);  
			if (add_iov(c, respH[nt].header.c_str(), respH[nt].header.length()) != 0
					|| add_iov(c,"0\r\n",3) != 0)
			{
				out_string(c, "ERROR!");
				return ;
			} 
			nt++;
		}  */
                 
		int hlen = sprintf(rbuf, " 0 %d\r\n", c->wbytes - (c->wcurr - c->wbuf) );
                string header = "VALUE ";
		header.append( keys[ result_idx ].c_str(), keys[ result_idx ].length());
		header.append(rbuf, hlen);  
		if (add_iov(c, header.c_str(), header.length()) != 0
				|| add_iov(c, c->wcurr, c->wbytes - (c->wcurr-c->wbuf)) != 0
				|| add_iov(c, "\r\n",2) != 0)
		{
			DEBUG4("write key=", MULTI_END_KEY , "; wbytes= ", c->wbytes);
			c->wbytes = 0;
			c->wcurr = 0;
		}
		DEBUG4("write key=", MULTI_END_KEY , "; wbytes= ", c->wbytes -(c->wcurr-c->wbuf));
		c->wcurr = c->wbuf +  c->wbytes;
               // nt++; 
	}
	else
	{

		DEBUG1("after Process, no data to write!");
		if(iret < 0)
			out_string(c, "ERROR!");
		else
			out_string(c, "NOTFOUND!");
	}
	/*
	   If the loop was terminated because of out-of-memory, it is not
	   reliable to add END\r\n to the buffer, because it might not end
	   in \r\n. So we send SERVER_ERROR instead.
	 */
	if (add_iov(c, "END\r\n", 5) != 0 || (c->udp && build_udp_headers(c) != 0)) {
		out_string(c, "SERVER_ERROR out of memory writing get response");
	}
	else {
		conn_set_state(c, conn_mwrite);
		c->msgcurr = 0;
	}
	return;
}


/* ntokens is overwritten here... shrug.. */
static inline void process_get_command(conn *c, token_t *tokens, size_t ntokens, bool return_cas) {
	char *key;
	size_t nkey;
	token_t *key_token = &tokens[KEY_TOKEN];
	assert(c != NULL);
        c->wbytes = 0;
        c->wcurr = c->wbuf;
        DEBUG2("ntockens  = ", ntokens ); 
	for(int i=0; i< (int) ntokens -1; i++)
		DEBUG4("tokens[", i, "] = ", tokens[i].value);

        respHeader_s respH[ntokens + 2];
        int nt = 0;
        Helper *p = Helper::Instance();        
	do {
		while(key_token->length != 0) {

			key = key_token->value;
			nkey = key_token->length;
                        DEBUG6("key=",key,"; nkey=",nkey,"; nt = ", nt);
			/*
			 * Construct the response. Each hit adds three elements to the
			 * outgoing data list:
			 *   "VALUE "
			 *   key
			 *   " " + flags + " " + data length + "\r\n" + data (with \r\n)
			 */

                        int iret = -1; 
			if( p != NULL )
			{
				DEBUG2( pthread_self(), " begin to Process");
				iret = p->Process((const char *)key, c);
				DEBUG2( pthread_self(), " end Process");
			}
			if( c->wbytes >(c->wcurr - c->wbuf) && iret == 0)
			{ 
				char rbuf[256];
				int hlen = sprintf(rbuf, " 0 %d\r\n", c->wbytes - (c->wcurr - c->wbuf) );
                                                                  
                                respH[nt].header.append("VALUE ");
                                respH[nt].header.append(key, nkey);
                                respH[nt].header.append(rbuf, hlen);  
                                
				if (add_iov(c, respH[nt].header.c_str(), respH[nt].header.length()) != 0
				   || add_iov(c, c->wcurr, c->wbytes - (c->wcurr-c->wbuf)) != 0
				   || add_iov(c, "\r\n",2) != 0)
				{
                                        DEBUG4("write key=", key, "; wbytes= ", c->wbytes);
                                        c->wbytes = 0;
                                        c->wcurr = 0;
					break;
				}
                                DEBUG4("write key=", key, "; value length ", c->wbytes - (c->wcurr - c->wbuf));
                                c->wcurr = c->wbuf +  c->wbytes;
			}
                        else
                        {
                           
                           DEBUG1("after Process, no data to write!");
                           if(iret < 0)
                              out_string(c, "ERROR!");
                           else
                              out_string(c, "NOTFOUND!");
                        }
			key_token++;
                        nt++;
		}

		/*
		 * If the command string hasn't been fully processed, get the next set
		 * of tokens.
		 */
		if(key_token->value != NULL) {
			ntokens = tokenize_command(key_token->value, tokens, MAX_TOKENS);
			key_token = tokens;
		}

	} while(key_token->value != NULL);

	/*
	   If the loop was terminated because of out-of-memory, it is not
	   reliable to add END\r\n to the buffer, because it might not end
	   in \r\n. So we send SERVER_ERROR instead.
	 */
	if (key_token->value != NULL || add_iov(c, "END\r\n", 5) != 0 || (c->udp && build_udp_headers(c) != 0)) {
		out_string(c, "SERVER_ERROR out of memory writing get response");
	}
	else {
		conn_set_state(c, conn_mwrite);
		c->msgcurr = 0;
	}
	return;
}

static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) {
//	char *key;
//	size_t nkey;
//	int flags;
//	time_t exptime;
	int vlen; //, old_vlen;
	assert(c != NULL);

	set_noreply_maybe(c, tokens, ntokens);

	if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) {
		out_string(c, "CLIENT_ERROR bad command line format");
		return;
	}

	c->key = tokens[KEY_TOKEN].value;
	c->nkey = tokens[KEY_TOKEN].length;
	c->flags = strtoul(tokens[2].value, NULL, 10);
	c->expire = strtol(tokens[3].value, NULL, 10);
	vlen = strtol(tokens[4].value, NULL, 10);
DEBUG2("cmd = ", c->key);
	// does cas value exist?
	if(errno == ERANGE ||  errno == EINVAL || vlen < 0) {
		out_string(c, "CLIENT_ERROR bad command line format");
		return;
	}
/*
	size_t t = 0;
	int voffset = 0;
	for(t =0 ; t < ntokens; t++)
	{
		voffset += (int) tokens[t].length;
		voffset ++;
	}
*/        
	c->vlen = vlen + 2;
	int rsize = (c->rcurr - c->rbuf) + c->offset + c->vlen;
	if( c->rsize < rsize )
	{
		char *new_rbuf =(char *) realloc(c->rbuf, rsize + 2);
		if (!new_rbuf) {
			fprintf(stderr, "Couldn't realloc input buffer\n");
			c->rbytes = 0; /* ignore what we read */
			out_string(c, "SERVER_ERROR out of memory reading request");
			c->write_and_go = conn_closing;
			return ;
		}
		c->rcurr = c->rbuf = new_rbuf;
		c->rsize = rsize + 2;
	}

	/* 此前，rcurr应该指向command的开始，voffset则是整个command的长度,此处使ritem指向数据区 */
	c->ritem = c->rcurr + c->offset; //     ITEM_data(it);
	/* rlbytes表示数据区还有多长数据没有读取！可能已经被读取完成！ */
	c->rlbytes =  (c->offset + c->vlen) - (c->rbytes - (c->rcurr - c->rbuf)); //        it->nbytes;
	conn_set_state(c, conn_nread);
}

static void process_command(conn *c, char *command) {

    token_t tokens[MAX_TOKENS];
    size_t ntokens;
    int comm;

    assert(c != NULL);

    //MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes);

    /*
     * for commands set/add/replace, we build an item and read the data
     * directly into it, then continue in nread_complete().
     */

    c->msgcurr = 0;
    c->msgused = 0;
    c->iovused = 0;
    if (add_msghdr(c) != 0) {
        out_string(c, "SERVER_ERROR out of memory preparing response");
        return;
    }

    ntokens = tokenize_command(command, tokens, MAX_TOKENS);
    DEBUG3(command, "; ntokens = ",  ntokens);

    for(unsigned i=0; i< ntokens-1; i++)
        DEBUG4("tokens[",i,"] = ", tokens[i].value );

    if (ntokens >= 3 && ((strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) ||
         (strcmp(tokens[COMMAND_TOKEN].value, "bget") == 0))) {
	    if( ntokens >= 5  && strcmp(tokens[ntokens - 2].value, MULTI_END_KEY ) == 0)
		    process_gets_command(c, tokens, ntokens, false);
	    else
		    process_get_command(c, tokens, ntokens, false);
    } else if ((ntokens == 6 || ntokens == 7) &&
               ((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) ||
                (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) ||
                (strcmp(tokens[COMMAND_TOKEN].value, "replace") == 0 && (comm = NREAD_REPLACE)) ||
                (strcmp(tokens[COMMAND_TOKEN].value, "prepend") == 0 && (comm = NREAD_PREPEND)) ||
                (strcmp(tokens[COMMAND_TOKEN].value, "append") == 0 && (comm = NREAD_APPEND)) )) {

        process_update_command(c, tokens, ntokens, comm, false);

    } else if (ntokens >= 3 && (strcmp(tokens[COMMAND_TOKEN].value, "gets") == 0)) {
	    if( ntokens >= 5  && strcmp(tokens[ntokens - 2].value, MULTI_END_KEY ) == 0)
		    process_gets_command(c, tokens, ntokens, true);
	    else
		    process_get_command(c, tokens, ntokens, true);
    } else if (ntokens >= 2 && (strcmp(tokens[COMMAND_TOKEN].value, "stats") == 0)) {
        process_stat(c, tokens, ntokens);
    } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "version") == 0)) {
        out_string(c, "VERSION " VERSION);
    } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "quit") == 0)) {
        conn_set_state(c, conn_closing);
    } else {
        out_string(c, "ERROR");
    }
    return;
}

/*
 * if we have a complete line in the buffer, process it.
 */
static int try_read_command(conn *c) {
    char *el, *cont;

    assert(c != NULL);
    assert(c->rcurr <= (c->rbuf + c->rsize));

    if (c->rbytes == 0)
        return 0;
    el = (char *)memchr(c->rcurr, '\n', c->rbytes);
    if (!el)
        return 0;
    cont = el + 1;
    if ((el - c->rcurr) > 1 && *(el - 1) == '\r') {
        el--;
    }
    *el = '\0';

    c->vlen = 0;
    assert(cont <= (c->rcurr + c->rbytes));
   // 这里rcurr首次校正，指向了命令的开始位置！
    c->offset = cont - c->rcurr;
    process_command(c, c->rcurr);

    // 命令解析完成后，校正rbytes,使其等于从数据区到rbuf中的当前读入数据结尾长度！
    c->rbytes -= c->offset;
   // 这里rcurr再次校正，指向了数据的开始位置！
    c->rcurr = cont;
    assert(c->rcurr <= (c->rbuf + c->rsize));

    return 1;
}

/*
 * read a UDP request.
 * return 0 if there's nothing to read.
 */
static int try_read_udp(conn *c) {
    int res;

    assert(c != NULL);

    c->request_addr_size = sizeof(c->request_addr);
    res = recvfrom(c->sfd, c->rbuf, c->rsize,
                   0, &c->request_addr, &c->request_addr_size);
    if (res > 8) {
        unsigned char *buf = (unsigned char *)c->rbuf;
        /* Beginning of UDP packet is the request ID; save it. */
        c->request_id = buf[0] * 256 + buf[1];

        /* If this is a multi-packet request, drop it. */
        if (buf[4] != 0 || buf[5] != 1) {
            out_string(c, "SERVER_ERROR multi-packet request not supported");
            return 0;
        }

        /* Don't care about any of the rest of the header. */
        res -= 8;
        memmove(c->rbuf, c->rbuf + 8, res);

        c->rbytes += res;
        c->rcurr = c->rbuf;
        return 1;
    }
    return 0;
}

/*
 * read from network as much as we can, handle buffer overflow and connection
 * close.
 * before reading, move the remaining incomplete fragment of a command
 * (if any) to the beginning of the buffer.
 * return 0 if there's nothing to read on the first read.
 */
static int try_read_network(conn *c) {
    int gotdata = 0;
    int res;

    assert(c != NULL);

    if (c->rcurr != c->rbuf) {
        if (c->rbytes != 0) /* otherwise there's nothing to copy */
            memmove(c->rbuf, c->rcurr, c->rbytes);
        c->rcurr = c->rbuf;
    }

    while (1) {
        if (c->rbytes >= c->rsize) {
            char *new_rbuf = (char *) realloc(c->rbuf, c->rsize * 2);
            if (!new_rbuf) {
                fprintf(stderr, "Couldn't realloc input buffer\n");
                c->rbytes = 0; /* ignore what we read */
                out_string(c, "SERVER_ERROR out of memory reading request");
                c->write_and_go = conn_closing;
                return 1;
            }
            c->rcurr = c->rbuf = new_rbuf;
            c->rsize *= 2;
        }

        /* unix socket mode doesn't need this, so zeroed out.  but why
         * is this done for every command?  presumably for UDP
         * mode.  */
        if (!settings.socketpath) {
            c->request_addr_size = sizeof(c->request_addr);
        } else {
            c->request_addr_size = 0;
        }

        int avail = c->rsize - c->rbytes;
        res = read(c->sfd, c->rbuf + c->rbytes, avail);
        if (res > 0) {
            gotdata = 1;
            c->rbytes += res;
            if (res == avail) {
                continue;
            } else {
                break;
            }
        }
        if (res == 0) {
            /* connection closed */
            conn_set_state(c, conn_closing);
            return 1;
        }
        if (res == -1) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) break;
            /* Should close on unhandled errors. */
            conn_set_state(c, conn_closing);
            return 1;
        }
    }
    return gotdata;
}

static bool update_event(conn *c, const int new_flags) {
    assert(c != NULL);

    struct event_base *base = c->event.ev_base;
    if (c->ev_flags == new_flags)
        return true;
    if (event_del(&c->event) == -1) return false;
    event_set(&c->event, c->sfd, new_flags, event_handler, (void *)c);
    event_base_set(base, &c->event);
    c->ev_flags = new_flags;
    if (event_add(&c->event, 0) == -1) return false;
    return true;
}

/*
 * Sets whether we are listening for new connections or not.
 */
void do_accept_new_conns(const bool do_accept) {
    conn *next;

    for (next = listen_conn; next; next = next->next) {
        if (do_accept) {
            update_event(next, EV_READ | EV_PERSIST);
            if (listen(next->sfd, settings.backlog) != 0) {
                perror("listen");
            }
        }
        else {
            update_event(next, 0);
            if (listen(next->sfd, 0) != 0) {
                perror("listen");
            }
        }
    }
//    if (do_accept) { } else {  }
}


/*
 * Transmit the next chunk of data from our list of msgbuf structures.
 *
 * Returns:
 *   TRANSMIT_COMPLETE   All done writing.
 *   TRANSMIT_INCOMPLETE More data remaining to write.
 *   TRANSMIT_SOFT_ERROR Can't write any more right now.
 *   TRANSMIT_HARD_ERROR Can't write (c->state is set to conn_closing)
 */
static int transmit(conn *c) {
    assert(c != NULL);

    if (c->msgcurr < c->msgused &&
            c->msglist[c->msgcurr].msg_iovlen == 0) {
        /* Finished writing the current msg; advance to the next. */
        c->msgcurr++;
    }
    if (c->msgcurr < c->msgused) {
        ssize_t res;
        struct msghdr *m = &c->msglist[c->msgcurr];

        res = sendmsg(c->sfd, m, 0);
        if (res > 0) {

            /* We've written some of the data. Remove the completed
               iovec entries from the list of pending writes. */
            while (m->msg_iovlen > 0 && res >= (ssize_t) m->msg_iov->iov_len) {
                res -= m->msg_iov->iov_len;
                m->msg_iovlen--;
                m->msg_iov++;
            }

            /* Might have written just part of the last iovec entry;
               adjust it so the next write will do the rest. */
            if (res > 0) {
                char *base = (char *) m->msg_iov->iov_base;
                m->msg_iov->iov_base = (void *)( base + res);
                m->msg_iov->iov_len -= res;
            }
            return TRANSMIT_INCOMPLETE;
        }
        if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
            if (!update_event(c, EV_WRITE | EV_PERSIST)) {
                    fprintf(stderr, "Couldn't update event\n");
                conn_set_state(c, conn_closing);
                return TRANSMIT_HARD_ERROR;
            }
            return TRANSMIT_SOFT_ERROR;
        }
        /* if res==0 or res==-1 and error is not EAGAIN or EWOULDBLOCK,
           we have a real error, on which we close the connection */
         perror("Failed to write, and not due to blocking");

        if (c->udp)
            conn_set_state(c, conn_read);
        else
            conn_set_state(c, conn_closing);
        return TRANSMIT_HARD_ERROR;
    } else {
        return TRANSMIT_COMPLETE;
    }
}

static void drive_machine(conn *c) {
    bool stop = false;
    int sfd, flags = 1;
    socklen_t addrlen;
    struct sockaddr_storage addr;
    int nreqs = settings.reqs_per_event;
    int res;

    assert(c != NULL);
    clientAcl *pAcl = clientAcl::Instance();
    while (!stop) {
        switch(c->state) {
        case conn_listening:
            addrlen = sizeof(addr);
            if ((sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen)) == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    /* these are transient, so don't log anything */
                    stop = true;
                } else if (errno == EMFILE) {
                        fprintf(stderr, "Too many open connections\n");
                    accept_new_conns(false);
                    stop = true;
                } else {
                    perror("accept()");
                    stop = true;
                }
                break;
            }

	    if(pAcl != NULL)
	    {
		    if( !pAcl->Check(&addr))
		    {
			    close(sfd);
			    ERRO   << "ip not in ACL list,can't accept it : " 
				    << inet_ntoa(((struct sockaddr_in *)(&addr))->sin_addr)  <<endl;
			    stop = true;
			    break;
		    } 
	    }
            else
               DEBUG1("ACL is NULL !" );
	    /* Accecc control module end*/

	    if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 ||
                fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
                perror("setting O_NONBLOCK");
                close(sfd);
                break;
            }
            dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST,
                                     DATA_BUFFER_SIZE, false);
            break;

        case conn_read:
            if (try_read_command(c) != 0) {
                continue;
            }
            /* Only process nreqs at a time to avoid starving other
               connections */
            if (--nreqs && (c->udp ? try_read_udp(c) : try_read_network(c)) != 0) {
                continue;
            }
            /* we have no command line and no data to read from network */
            if (!update_event(c, EV_READ | EV_PERSIST)) {
                    fprintf(stderr, "Couldn't update event\n");
                conn_set_state(c, conn_closing);
                break;
            }
            stop = true;
            break;

        case conn_nread:
            /* we are reading rlbytes into ritem; */
            if (c->rlbytes <= 0) {
                complete_nread(c);
                break;
            }
            /* first check if we have leftovers in the conn_read buffer */
      
            /*  now try reading from the socket */
            /* 在process_update_command中已经把ritem指向了数据起始位置，
               try_read_command中把rbytes修正为从ritem之后的已读数据长度,
               process_update_command把 rlbytes调整为等待读取的数据长度               
            */
            res = read(c->sfd, c->ritem + c->rbytes, c->rlbytes);
            if (res > 0) {
             //   c->ritem += res;
                c->rbytes += res;
                c->rlbytes -= res;
                break;
            }
            if (res == 0) { /* end of stream */
                conn_set_state(c, conn_closing);
                break;
            }
            if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
                if (!update_event(c, EV_READ | EV_PERSIST)) {
                        fprintf(stderr, "Couldn't update event\n");
                    conn_set_state(c, conn_closing);
                    break;
                }
                stop = true;
                break;
            }
            /* otherwise we have a real error, on which we close the connection */
                fprintf(stderr, "Failed to read, and not due to blocking\n");
            conn_set_state(c, conn_closing);
            break;

        case conn_swallow:
            /* we are reading sbytes and throwing them away */
            if (c->sbytes == 0) {
                conn_set_state(c, conn_read);
                break;
            }

            /* first check if we have leftovers in the conn_read buffer */
            if (c->rbytes > 0) {
                int tocopy = c->rbytes > c->sbytes ? c->sbytes : c->rbytes;
                c->sbytes -= tocopy;
                c->rcurr += tocopy;
                c->rbytes -= tocopy;
                break;
            }

            /*  now try reading from the socket */
            res = read(c->sfd, c->rbuf, c->rsize > c->sbytes ? c->sbytes : c->rsize);
            if (res > 0) {
                c->sbytes -= res;
                break;
            }
            if (res == 0) { /* end of stream */
                conn_set_state(c, conn_closing);
                break;
            }
            if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
                if (!update_event(c, EV_READ | EV_PERSIST)) {
                        fprintf(stderr, "Couldn't update event\n");
                    conn_set_state(c, conn_closing);
                    break;
                }
                stop = true;
                break;
            }
            /* otherwise we have a real error, on which we close the connection */
                fprintf(stderr, "Failed to read, and not due to blocking\n");
            conn_set_state(c, conn_closing);
            break;

        case conn_write:
            /*
             * We want to write out a simple response. If we haven't already,
             * assemble it into a msgbuf list (this will be a single-entry
             * list for TCP or a two-entry list for UDP).
             */
            if (c->iovused == 0 || (c->udp && c->iovused == 1)) {
                if (add_iov(c, c->wcurr, c->wbytes) != 0 ||
                    (c->udp && build_udp_headers(c) != 0)) {
                        fprintf(stderr, "Couldn't build response\n");
                    conn_set_state(c, conn_closing);
                    break;
                }
            }

            /* fall through... */

        case conn_mwrite:
            switch (transmit(c)) {
            case TRANSMIT_COMPLETE:
                conn_wbuf_shrink(c);  
                if (c->state == conn_mwrite) {
                    conn_set_state(c, conn_read);
                } else if (c->state == conn_write) {
                    if (c->write_and_free) {
                        free(c->write_and_free);
                        c->write_and_free = 0;
                    }
                    conn_set_state(c, c->write_and_go);
                } else {
                        fprintf(stderr, "Unexpected state %d\n", c->state);
                    conn_set_state(c, conn_closing);
                }
                break;

            case TRANSMIT_INCOMPLETE:
            case TRANSMIT_HARD_ERROR:
                break;                   /* Continue in state machine. */

            case TRANSMIT_SOFT_ERROR:
                stop = true;
                break;
            }
            break;

        case conn_closing:
            if (c->udp)
                conn_cleanup(c);
            else
                conn_close(c);
            stop = true;
            break;
        }
    }

    return;
}

void event_handler(const int fd, const short which, void *arg) {
    conn *c;

    c = (conn *)arg;
    assert(c != NULL);

    c->which = which;

    /* sanity */
    if (fd != c->sfd) {
            fprintf(stderr, "Catastrophic: event fd doesn't match conn fd!\n");
        conn_close(c);
        return;
    }

    drive_machine(c);

    /* wait for next event */
    return;
}

static int new_socket(struct addrinfo *ai) {
    int sfd;
    int flags;

    if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) {
        return -1;
    }

    if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 ||
        fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
        perror("setting O_NONBLOCK");
        close(sfd);
        return -1;
    }
    return sfd;
}


/*
 * Sets a socket's send buffer size to the maximum allowed by the system.
 */
static void maximize_sndbuf(const int sfd) {
    socklen_t intsize = sizeof(int);
    int last_good = 0;
    int min, max, avg;
    int old_size;

    /* Start with the default size. */
    if (getsockopt(sfd, SOL_SOCKET, SO_SNDBUF, &old_size, &intsize) != 0) {
            perror("getsockopt(SO_SNDBUF)");
        return;
    }

    /* Binary-search for the real maximum. */
    min = old_size;
    max = MAX_SENDBUF_SIZE;

    while (min <= max) {
        avg = ((unsigned int)(min + max)) / 2;
        if (setsockopt(sfd, SOL_SOCKET, SO_SNDBUF, (void *)&avg, intsize) == 0) {
            last_good = avg;
            min = avg + 1;
        } else {
            max = avg - 1;
        }
    }
    DEBUG5(sfd, " :  send buffer was ", old_size, ", now is ", last_good);
}

static int server_socket(const int port, const bool is_udp) {
    int sfd;
    struct linger ling = {0, 0};
    struct addrinfo *ai;
    struct addrinfo *next;
    struct addrinfo hints;
    char port_buf[NI_MAXSERV];
    int error;
    int success = 0;

    int flags =1;

    /*
     * the memset call clears nonstandard fields in some impementations
     * that otherwise mess things up.
     */
    memset(&hints, 0, sizeof (hints));
    hints.ai_flags  = AI_PASSIVE;
    hints.ai_family = AF_UNSPEC;
    if (is_udp)
    {
        hints.ai_socktype = SOCK_DGRAM;
    } else {
        hints.ai_socktype = SOCK_STREAM;
    }

    snprintf(port_buf, NI_MAXSERV, "%d", port);
    error= getaddrinfo(settings.inter, port_buf, &hints, &ai);
    if (error != 0) {
        if (error != EAI_SYSTEM)
            fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error));
        else
            perror("getaddrinfo()");

        return 1;
    }

    for (next= ai; next; next= next->ai_next) {
        conn *listen_conn_add;
        if ((sfd = new_socket(next)) == -1) {
            /* getaddrinfo can return "junk" addresses,
             * we make sure at least one works before erroring.
             */
            continue;
        }

#ifdef IPV6_V6ONLY
        if (next->ai_family == AF_INET6) {
            error = setsockopt(sfd, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &flags, sizeof(flags));
            if (error != 0) {
                perror("setsockopt");
                close(sfd);
                continue;
            }
        }
#endif

        setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags));
        if (is_udp) {
            maximize_sndbuf(sfd);
        } else {
            error = setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags));
            if (error != 0)
                perror("setsockopt");

            error = setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling));
            if (error != 0)
                perror("setsockopt");

            error = setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags));
            if (error != 0)
                perror("setsockopt");
        }

        if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) {
            if (errno != EADDRINUSE) {
                perror("bind()");
                close(sfd);
                freeaddrinfo(ai);
                return 1;
            }
            close(sfd);
            continue;
        } else {
            success++;
            if (!is_udp && listen(sfd, settings.backlog) == -1) {
                perror("listen()");
                close(sfd);
                freeaddrinfo(ai);
                return 1;
            }
        }

        if (is_udp)
        {
          int c;

          for (c = 1; c < settings.num_threads; c++) {
              /* this is guaranteed to hit all threads because we round-robin */
              dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST,
                                UDP_READ_BUFFER_SIZE, is_udp);
          }
        } else {
          if (!(listen_conn_add = conn_new(sfd, conn_listening,
                                           EV_READ | EV_PERSIST, 1, false, main_base))) {
              fprintf(stderr, "failed to create listening connection\n");
              exit(EXIT_FAILURE);
          }

          listen_conn_add->next = listen_conn;
          listen_conn = listen_conn_add;
        }
    }

    freeaddrinfo(ai);

    /* Return zero iff we detected no errors in starting up connections */
    return success == 0;
}

static int new_socket_unix(void) {
    int sfd;
    int flags;

    if ((sfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
        perror("socket()");
        return -1;
    }

    if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 ||
        fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
        perror("setting O_NONBLOCK");
        close(sfd);
        return -1;
    }
    return sfd;
}

static int server_socket_unix(const char *path, int access_mask) {
    int sfd;
    struct linger ling = {0, 0};
    struct sockaddr_un addr;
    struct stat tstat;
    int flags =1;
    int old_umask;

    if (!path) {
        return 1;
    }

    if ((sfd = new_socket_unix()) == -1) {
        return 1;
    }

    /*
     * Clean up a previous socket file if we left it around
     */
    if (lstat(path, &tstat) == 0) {
        if (S_ISSOCK(tstat.st_mode))
            unlink(path);
    }

    setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags));
    setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags));
    setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling));

    /*
     * the memset call clears nonstandard fields in some impementations
     * that otherwise mess things up.
     */
    memset(&addr, 0, sizeof(addr));

    addr.sun_family = AF_UNIX;
    strcpy(addr.sun_path, path);
    old_umask=umask( ~(access_mask&0777));
    if (bind(sfd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
        perror("bind()");
        close(sfd);
        umask(old_umask);
        return 1;
    }
    umask(old_umask);
    if (listen(sfd, settings.backlog) == -1) {
        perror("listen()");
        close(sfd);
        return 1;
    }
    if (!(listen_conn = conn_new(sfd, conn_listening,
                                     EV_READ | EV_PERSIST, 1, false, main_base))) {
        fprintf(stderr, "failed to create listening connection\n");
        exit(EXIT_FAILURE);
    }

    return 0;
}

/*
 * We keep the current time of day in a global variable that's updated by a
 * timer event. This saves us a bunch of time() system calls (we really only
 * need to get the time once a second, whereas there can be tens of thousands
 * of requests a second) and allows us to use server-start-relative timestamps
 * rather than absolute UNIX timestamps, a space savings on systems where
 * sizeof(time_t) > sizeof(unsigned int).
 */
static struct event clockevent;
/* time-sensitive callers can call it by hand with this, outside the normal ever-1-second timer */

static void clock_handler(const int fd, const short which, void *arg) {
    struct timeval t = { settings.timer,  0};
    static bool initialized = false;

    if (initialized) {
        /* only delete the event if it's actually there. */
        evtimer_del(&clockevent);
    } else {
        initialized = true;
    }

    evtimer_set(&clockevent, clock_handler, 0);
    event_base_set(main_base, &clockevent);
    evtimer_add(&clockevent, &t);

    Helper *p = Helper::Instance();
    if(p != NULL)
          p->Timer(time(NULL), settings.timer);
    // do something !
}

static void usage(void) {
    printf(PACKAGE " " VERSION "\n");
    printf("-p <num>      TCP port number to listen on (default: 11211)\n"
           "-z <num>      zexmon port number to listen on (default: 0 is off)\n"
           "-U <num>      UDP port number to listen on (default: 0 is off)\n"
           "-s <file>     unix socket path to listen on (disables network support)\n"
           "-a <mask>     access mask for unix socket, in octal (default 0700)\n"
           "-l <ip_addr>  interface to listen on, default is INDRR_ANY\n"
           "-d            run as a daemon\n"
           "-r            maximize core file limit\n"
           "-u <username> assume identity of <username> (only when run as root)\n"
           "-c <num>      max simultaneous connections, default is 1024\n"
           "-h            print this help and exit\n"
           "-i            print memcached and libevent license\n"
           "-P <file>     save PID in <file>, only used with -d option\n"
           "-f <file>     IP acl cnofigure <file>\n"
           "-m <file>     helper configure <file>, default is ./helper.conf\n"
           "-t <num>      timer event , default 0\n"
           );

    printf("-n <num>      number of threads to use, default 4\n");
    printf("-R            Maximum number of requests per event\n"
           "              limits the number of requests process for a given con nection\n"
           "              to prevent starvation.  default 20\n");
    printf("-b            Set the backlog queue limit (default 1024)\n");
    return;
}

static void usage_license(void) {
    printf(PACKAGE " " VERSION "\n\n");
    printf(
    "Copyright (c) 2003, Danga Interactive, Inc. <http://www.danga.com/>\n"
    "All rights reserved.\n"
    );
    return;
}

static void save_pid(const pid_t pid, const char *pid_file) 
{
	FILE *fp;
	if (pid_file == NULL)
		return;

	if ((fp = fopen(pid_file, "w")) == NULL) {
		fprintf(stderr, "Could not open the pid file %s for writing\n", pid_file);
		return;
	}

	fprintf(fp,"%ld\n", (long)pid);
	if (fclose(fp) == -1) {
		fprintf(stderr, "Could not close the pid file %s.\n", pid_file);
		return;
	}
}

static void remove_pidfile(const char *pid_file) 
{
	if (pid_file == NULL)
		return;

	if (unlink(pid_file) != 0) {
		fprintf(stderr, "Could not remove the pid file %s.\n", pid_file);
	}

}


static void sig_handler(const int sig) 
{
    printf("SIGINT handled.\n");
    exit(EXIT_SUCCESS);
}

#ifdef LINUX
   #define POSIXLY "+"
#else
   #define POSIXLY ""
#endif

int main (int argc, char **argv) {
    int c;
//    int x;
    bool daemonize = false;
    int maxcore = 0;
    char *username = NULL;
    char *pid_file = NULL;
    struct passwd *pw;
    struct sigaction sa;
    struct rlimit rlim;
    /* listening socket */
    static int *l_socket = NULL;

    /* udp socket */
    static int *u_socket = NULL;
    uptime = time(NULL);
    /* handle SIGINT */
    signal(SIGINT, sig_handler);
    /* init settings */
    settings_init();
    /* set stderr non-buffering (for running under, say, daemontools) */
    setbuf(stderr, NULL);

    /* process arguments */
    while ((c = getopt(argc, argv, POSIXLY"a:p:s:U:m:Mc:khirvdl:u:P:f:s:z:w:n:t:D:LR:b:o:")) != -1) {
        switch (c) {
        case 'z':
                zexmon_port = atoi(optarg);
                break;
        case 'w':
                report_port = atoi(optarg);
                break;
        case 'a':
            /* access for unix domain socket, as octal mask (like chmod)*/
            settings.access= strtol(optarg,NULL,8);
            break;
        case 'U':
            settings.udpport = atoi(optarg);
            break;
        case 'p':
            settings.port = atoi(optarg);
            break;
        case 's':
            settings.socketpath = optarg;
            break;
        case 'm':
            settings.helper = optarg; 
            break;
        case 'c':
            settings.maxconns = atoi(optarg);
            break;
        case 'h':
            usage();
            exit(EXIT_SUCCESS);
        case 'i':
            usage_license();
            exit(EXIT_SUCCESS);
        case 'v':

            break;
        case 'l':
            settings.inter= strdup(optarg);
            break;
        case 'd':
            daemonize = true;
            break;
        case 'r':
            maxcore = 1;
            break;
        case 'R':
            settings.reqs_per_event = atoi(optarg);
            if (settings.reqs_per_event == 0) {
                fprintf(stderr, "Number of requests per event must be greater than 0\n");
                return 1;
            }
            break;
        case 'u':
            username = optarg;
            break;
        case 'P':
            pid_file = optarg;
            break;
        case 'f':
	    settings.authfile = strdup(optarg);
            break;
        case 'n':
            settings.num_threads = atoi(optarg) + 1; /* Extra dispatch thread */
            if (settings.num_threads < 2) {
                fprintf(stderr, "Number of threads must be greater than 0\n");
                return 1;
            }
            break;
        case 't':
           settings.timer = abs(atoi(optarg));
           break;
        case 'D':
            settings.detail_enabled = 1;
            break;
        case 'b' :
            settings.backlog = atoi(optarg);
            break;
        default:
            fprintf(stderr, "Illegal argument \"%c\"\n", c);
            return 1;
        }
    }

    clientAcl *acl =  clientAcl::Instance(settings.authfile);
    if(acl != NULL)
    {
	    if(report_port > 0 )
		    acl->Report_Restart( report_port, settings.port);
	    acl->Print();
    }
    else
       DEBUG1("ACL IS NULL!");

    Helper *hp =  Helper::Instance(settings.helper);
    if(hp == NULL)
    {
          ERRO <<"Can't Helper::Instance"<<endl;
          exit(0);
    }

    if (maxcore != 0) 
    {
	    struct rlimit rlim_new;
	    /*
	     * First try raising to infinity; if that fails, try bringing
	     * the soft limit to the hard.
	     */
	    if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
		    rlim_new.rlim_cur = rlim_new.rlim_max = RLIM_INFINITY;
		    if (setrlimit(RLIMIT_CORE, &rlim_new)!= 0) {
			    /* failed. try raising just to the old max */
			    rlim_new.rlim_cur = rlim_new.rlim_max = rlim.rlim_max;
			    (void)setrlimit(RLIMIT_CORE, &rlim_new);
		    }
	    }
	    /*
	     * getrlimit again to see what we ended up with. Only fail if
	     * the soft limit ends up 0, because then no core files will be
	     * created at all.
	     */

	    if ((getrlimit(RLIMIT_CORE, &rlim) != 0) || rlim.rlim_cur == 0) {
		    fprintf(stderr, "failed to ensure corefile creation\n");
		    exit(EXIT_FAILURE);
	    }
    }

    /*
     * If needed, increase rlimits to allow as many connections
     * as needed.
     */

    if (getrlimit(RLIMIT_NOFILE, &rlim) != 0) {
	    fprintf(stderr, "failed to getrlimit number of files\n");
	    exit(EXIT_FAILURE);
    } else {
	    int maxfiles = settings.maxconns;
	    if (rlim.rlim_cur < (size_t) maxfiles)
		    rlim.rlim_cur = maxfiles + 3;
	    if (rlim.rlim_max < rlim.rlim_cur)
		    rlim.rlim_max = rlim.rlim_cur;
	    if (setrlimit(RLIMIT_NOFILE, &rlim) != 0) {
		    fprintf(stderr, "failed to set rlimit for open files. Try running as root or requesting smaller maxconns value.\n");
		    exit(EXIT_FAILURE);
	    }
    }

    /* daemonize if requested */
    /* if we want to ensure our ability to dump core, don't chdir to / */
    if (daemonize) {
        int res;
        res = daemon(maxcore, 0); 
        if (res == -1) {
            fprintf(stderr, "failed to daemon() in order to daemonize\n");
            return 1;
        }
    }

    /* lose root privileges if we have them */
    if (getuid() == 0 || geteuid() == 0) {
        if (username == 0 || *username == '\0') {
            fprintf(stderr, "can't run as root without the -u switch\n");
            return 1;
        }
        if ((pw = getpwnam(username)) == 0) {
            fprintf(stderr, "can't find the user %s to switch to\n", username);
            return 1;
        }
        if (setgid(pw->pw_gid) < 0 || setuid(pw->pw_uid) < 0) {
            fprintf(stderr, "failed to assume identity of user %s\n", username);
            return 1;
        }
    }

    /*
     * Add by lengss for zexmon
     */
    if(zexmon_port > 0)
    {
	    pthread_t       exmonthread;
	    pthread_attr_t  attr;
	    int             ret;
	    pthread_attr_init(&attr);
	    if((ret = pthread_create(&exmonthread, &attr, zexmon_command_waiter , &zexmon_port)) != 0)
	    {
		    fprintf(stderr, "Can't create thread: %s\n", strerror(ret));
		    exit(1);
	    }
    }

    /* initialize main thread libevent instance */
    main_base = event_init();
    conn_init();

    /*
     * ignore SIGPIPE signals; we can use errno==EPIPE if we
     * need that information
     */
    sa.sa_handler = SIG_IGN;
    sa.sa_flags = 0;
    if (sigemptyset(&sa.sa_mask) == -1 ||
        sigaction(SIGPIPE, &sa, 0) == -1) {
        perror("failed to ignore SIGPIPE; sigaction");
        exit(EXIT_FAILURE);
    }
    /* start up worker threads if MT mode */
    thread_init(settings.num_threads, main_base);
    /* save the PID in if we're a daemon, do this after thread_init due to
       a file descriptor handling bug somewhere in libevent */
    if (daemonize)
        save_pid(getpid(), pid_file);
    /* initialise clock event */
    if(settings.timer > 0)
	    clock_handler(0, 0, 0);
    /* initialise deletion array and timer event */
   
 /* create unix mode sockets after dropping privileges */
    if (settings.socketpath != NULL) {
        errno = 0;
        if (server_socket_unix(settings.socketpath,settings.access)) {
          fprintf(stderr, "failed to listen on UNIX socket: %s\n", settings.socketpath);
          if (errno != 0)
              perror("socket listen");
          exit(EXIT_FAILURE);
        }
    }

    /* create the listening socket, bind it, and init */
    if (settings.socketpath == NULL) {
        errno = 0;
        if (settings.port && server_socket(settings.port, 0)) {
            fprintf(stderr, "failed to listen on TCP port %d\n", settings.port);
            if (errno != 0)
                perror("tcp listen");
            exit(EXIT_FAILURE);
        }
        /*
         * initialization order: first create the listening sockets
         * (may need root on low ports), then drop root if needed,
         * then daemonise if needed, then init libevent (in some cases
         * descriptors created by libevent wouldn't survive forking).
         */

        /* create the UDP listening socket and bind it */
        errno = 0;
        if (settings.udpport && server_socket(settings.udpport, 1)) {
            fprintf(stderr, "failed to listen on UDP port %d\n", settings.udpport);
            if (errno != 0)
                perror("udp listen");
            exit(EXIT_FAILURE);
        }
    }

    pthread_t       checkerthread;
    pthread_attr_t  attr2;
    int             ret2;
    pthread_attr_init(&attr2);
    if((ret2 = pthread_create(&checkerthread, &attr2, extern_deamon_checker , &settings.port)) != 0)
    {
	    fprintf(stderr, "Can't create thread: %s\n", strerror(ret2));
	    exit(1);
    }

    /* enter the event loop */
    event_base_loop(main_base, 0);
    /* remove the PID file if we're a daemon */
    if (daemonize)
	    remove_pidfile(pid_file);
    /* Clean up strdup() call for bind() address */
    if (settings.inter)
	    free(settings.inter);
    if (l_socket)
	    free(l_socket);
    if (u_socket)
	    free(u_socket);
    return 0;
}
